package com.raylu.realtime.utils;

import com.raylu.realtime.bean.TransientSink;
import com.raylu.realtime.utils.PropertiesUtil;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.io.IOException;
import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Properties;

/**
 * Description:
 * <p>
 * Create by lucienoz on 2022/1/5.
 * Copyright © 2022 lucienoz. All rights reserved.
 */
public class ClickHouseUtil {

    public static <T> SinkFunction<T> getJdbcSink(String sql) throws IOException {
        Properties load = PropertiesUtil.load("config.properties");
        SinkFunction<T> sink = JdbcSink.<T>sink(sql, new JdbcStatementBuilder<T>() {
                    @Override
                    public void accept(PreparedStatement preparedStatement, T t) throws SQLException {
                        Field[] declaredFields = t.getClass().getDeclaredFields();
                        int noFieldCount = 0;

                        for (int i = 0; i < declaredFields.length; i++) {
                            Field declaredField = declaredFields[i];
                            declaredField.setAccessible(true);
                            TransientSink annotation = declaredField.getAnnotation(TransientSink.class);
                            if (annotation!=null) {
                                noFieldCount++;
                                continue;
                            }
                            try {
                                Object fieldValue = declaredField.get(t);
                                preparedStatement.setObject(i+1-noFieldCount,fieldValue);
                            } catch (IllegalAccessException e) {
                                e.printStackTrace();
                            }

                        }

                    }
                }, new JdbcExecutionOptions.Builder()
                        .withBatchSize(5)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName(load.getProperty("visitor.stats.app.clickhouse.driver"))
                        .withUrl(load.getProperty("visitor.stats.app.clickhouse.url"))
                        .build()
        );
        return sink;
    }


}
